package midas.mw.crt;

import midas.mw.crt.common.CBMessage;

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Collection;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Vector;

import midas.mw.cfg.CfgLogging;

public class CBRCentralNodeServer implements Runnable{

	private static CbrImpl cbrInterface = null;
	private static final boolean debug = true;
	ServerSocket serverSocket = null;
	Socket clientSocket = null;
	int MAX_PACKET_LEN = 2000;
	int usedPort;
	String usedIP;
	long msg_timeout_ms = 60 * 1000; /* 60s */ 
	boolean runCondition = true;
	QueueEmulate msgQueueRef;
	//Hashtable<CBMessage, LinkedList<Integer>> receivedBroadcastMsgs;
	Hashtable receivedBroadcastMsgs;

	public CBRCentralNodeServer(CbrImpl cbrInp, int port, String ip, QueueEmulate msgQueue){
		cbrInterface = cbrInp;
		usedPort = port;
		usedIP = ip;
		this.msgQueueRef = msgQueue;
		initServer(usedPort,usedIP);
		//receivedBroadcastMsgs = new Hashtable<CBMessage, LinkedList<Integer>>();
		receivedBroadcastMsgs = new Hashtable();
	}

	public static CbrImpl getCbrHandle()
	{
		return cbrInterface;
	}

	private void initServer(int port, String ip){
		try {
			serverSocket = new ServerSocket(port,0,Inet4Address.getByName(ip));
		} catch (IOException e) {
			CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_ERROR, "initServer(): SocketException " + e);
		}

		if (serverSocket==null){
			CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_INFORMAL, "Failed to start CBRCNServer !!!");
		}
	}
	
	public void run(){

		if (serverSocket==null){
			return;
		}
		byte[][] serverResponse = null;

		int requestPort;
		InetAddress requestAddress;
		byte[] firstMsg;
		DataInputStream in = null;
		DataOutputStream out = null;
		try{
			CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_INFORMAL, "CBR CN Server started on port "+serverSocket.getLocalPort());
			while (runCondition){
				byte[] buffer = new byte[MAX_PACKET_LEN];
				//byte[] buffer = new byte[MAX_PACKET_LEN];
				//request = new DatagramPacket(buffer,buffer.length);
				//serverSocket.receive(request);
				clientSocket = serverSocket.accept();
				in = new DataInputStream( clientSocket.getInputStream());
				out = new DataOutputStream( clientSocket.getOutputStream());

				CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "Client connected, read message");
			    
				/*
				int msglen = in.read(buffer,0,MAX_PACKET_LEN);
			    byte[] req_b = new byte[msglen];
			    System.arraycopy(buffer, 0, req_b, 0, msglen);
			    */
				byte[] receivedMessage = CBRCentralNodeClient.readPacket(in);
				CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "Message received " + new String(receivedMessage));  				
				serverResponse = generateResponse(receivedMessage);

				int len = 0;
				for (; len < serverResponse.length; len++){
					if (serverResponse[len] == null)
						break;
				}
 
				firstMsg = ("SIZE=" + len + "#").getBytes();
				CBRCentralNodeClient.writePacket(firstMsg, out);
				//out.write(makeDatagram(firstMsg));			
				
				if (debug){
					CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "Response sent " + new String(firstMsg));
				}

				for (int i = 0; i < len; i++){
					CBRCentralNodeClient.writePacket(serverResponse[i], out);
//					out.write(makeDatagram(serverResponse[i]));		
					if (debug){
						CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "Response sent " + serverResponse[i]);
					}
				}
				
				in.close();
				out.close();
				clientSocket.close();
				
			}
			CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_INFORMAL, "CBR CN Server stopped.");
		} catch (IOException ioe) {
			CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_ERROR, "runServer(): IOException " + ioe);
		}
	}
	/*
	byte[] makeDatagram(byte[]inp){
		byte[] bufferb = new byte[MAX_PACKET_LEN];
		for (int i=0;i<MAX_PACKET_LEN;i++) bufferb[i] = 0; 
		System.arraycopy(inp, 0, bufferb, 0, inp.length);
		return bufferb;
	}
	*/
	private byte[][] generateResponseTest(byte[] inp_b){
		
		byte[][] a = {"AAAAaaaaaaaaaaaa1".getBytes(),"BBBBbb2".getBytes(), "CCCC@@*&@)($&_(@*$(  _(@*@* @ @ @ \n \r \0 qaaaaaaaa3".getBytes()};
		return a;
	}
	
	private byte[][] generateResponse(byte[] inp_b)
	{
		Integer id = new Integer( ((int)inp_b[0] << 24) | ((int)inp_b[1] << 16) | ((int)inp_b[2] << 8) | (int)inp_b[3] );
		byte[] b = new byte[inp_b.length - 4];

		System.arraycopy(inp_b, 4, b, 0, inp_b.length - 4);

		CBMessage inp_msg = new CBMessage("-2", b, 0);

		CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "-> CBR message from " + id + " = " + inp_msg);
		CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, " destination: " + inp_msg.getDestination());
		CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, " payload: " + new String(b));

		((CbrImpl)getCbrHandle()).putCBMessage(inp_msg);

		int s = msgQueueRef.size(), r = 0;
		byte[][] responses = new byte[s][];
		int cur_idx = 0;

		for (int i = 0; i < s; i++) {
			CBMessage msg;

			CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg queue size: " + s + " loop nr: " + i);

			responses[i] = null;

			try {
				msg = (CBMessage)msgQueueRef.get(cur_idx);

				CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "<- CBR message hash = " + msg.hashCode());
				CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, " destination: " + msg.getDestination());
				CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, " payload: " + new String(msg.getPayload()));

				// handle broadcast messages first
				if (msg.getDestination().compareTo("-1") == 0) {
					CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg matches (broadcast)");

					if (System.currentTimeMillis() - msg.getTimestamp() < msg_timeout_ms) {
						//LinkedList<Integer> l;
						LinkedList l;

						// check if message was already received by this node
						if (receivedBroadcastMsgs.containsKey(msg)) {
							CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg already in hash");
							l = (LinkedList) receivedBroadcastMsgs.get(msg);
						} else {
							CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg added to hash");
							//l = new LinkedList<Integer>();
							l = new LinkedList();							
							receivedBroadcastMsgs.put(msg, l);
						}

						if (l.contains(id)) {
							CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "id already in hash");
						} else {
							// mark that the message was received by this node
							CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "id added to hash");
							l.add(id);

							responses[r] = new byte[msg.getPayload().length];
							System.arraycopy(msg.getPayload(), 0, responses[r], 0, msg.getPayload().length);
							r++;
						}

						cur_idx++;
					} else {
						CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg timed out");

						if (receivedBroadcastMsgs.containsKey(msg)) {
							receivedBroadcastMsgs.remove(msg);
							CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg removed");
						}

						msgQueueRef.dequeue(cur_idx);
					}
				} else if (msg.getDestination().compareTo(id.toString()) == 0) {
					CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg matches");
					responses[r] = new byte[msg.getPayload().length];
					System.arraycopy(msg.getPayload(), 0, responses[r], 0, msg.getPayload().length);
					r++;

					msgQueueRef.dequeue(cur_idx);
				} else {
					CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg doesn't match");

					if (System.currentTimeMillis() - msg.getTimestamp() < msg_timeout_ms) {
						cur_idx++;
					} else {
						CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "msg to (" + msg.getDestination() + ") timed out");
						msgQueueRef.dequeue(cur_idx);
					}
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		CfgLogging.logEvent(CfgLogging.LOG_INFO, "CRT.CbrCNServer", CfgLogging.SEV_DEBUG, "response ready");

		return responses;
	}
	
	public void stop(){
		runCondition = false;
		if (serverSocket!=null){
			try {
				serverSocket.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
}
